package com.xjy.rpc.netty;

import io.netty.channel.DefaultEventLoop;
import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import net.bytebuddy.implementation.bytecode.Throw;

import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * @description:
 * jdk : future,CompletableFuture,futureTask  netty : future,promise 对比
 * 同步、异步 针对 用户端调用        排队打饭和定外卖取餐
 * 阻塞、非阻塞 针对 cpu任务调度     饭没做好，被迫排队，找个人(Listener)饭好了通知
 * @author XJY
 * @date 2024/9/30 10:57
 * @version 1.0
 */
public class PromiseTest {

    public static void main(String[] args) throws Exception {
        PromiseTest test = new PromiseTest();
        //jdk这三个执行会阻塞，get方法是阻塞方法，单线程就是同步阻塞的
        test.jdkFuture();
        test.jdkCompletableFuture();
        test.jdkFutureTask();

        test.nettyFuture();
        test.nettyPromise();
    }

    /**
     * @description: TODO jdk future 演示
     * @author XJY
     * @date 2024/10/9 9:09
     * @version 1.0
     */
    public void jdkFuture() throws ExecutionException, InterruptedException, TimeoutException {
        ExecutorService pool = Executors.newCachedThreadPool();

        Future<String> future = pool.submit(
                () -> {
                    System.out.println("异步任务开始执行");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return "success";
                }
        );
        //异步获取结果
        String result = future.get(10,TimeUnit.SECONDS);
        //String result = future.get();
        System.out.println("jdkFuture : " + result);
        pool.shutdown();
    }

    /**
     * @description: TODO jdkCompletableFuture 的使用和链式写法
     * @author XJY
     * @date 2024/10/9 9:55
     * @version 1.0
     */
    public void jdkCompletableFuture() throws ExecutionException, InterruptedException {
        //CompletableFuture可指定线程池执行异步任务
        ExecutorService pool = Executors.newFixedThreadPool(2);
        //TODO 这里改为链式写法
        /*CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                System.out.println("异步任务开始执行");
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                return "success";
            }
        }, pool);
        String result = future.get();
        System.out.println("jdkComFuture : " + result);*/
        CompletableFuture.supplyAsync(() -> {
                    System.out.println("异步任务开始执行");
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return "Success";
                }).//处理结果
                thenApplyAsync(result -> result + ": 结果处理").//结果输出
                thenAcceptAsync(s -> System.out.println("jdkComFuture : " +s)).get();
        pool.shutdown();
    }

    /**
     * @description: jdkFutureTask 是Future的实现类，可创建任务交由线程池或异步线程执行，用get获取结果
     * @author XJY
     * @date 2024/10/9 10:03
     * @version 1.0
     */
    public void jdkFutureTask() throws ExecutionException, InterruptedException {
        FutureTask<String> futureTask = new FutureTask<>(
                () -> {
                    System.out.println("异步任务开始执行");
                    Thread.sleep(3000);
                    return "Success";
                }
        );

        ExecutorService pool = Executors.newCachedThreadPool();
        pool.submit(futureTask);

        String result = futureTask.get();
        System.out.println("jdkFutureTask : "+ result);
        pool.shutdown();
    }

    /**
     * @description: netty future 对jdk future的增强
     * @author XJY
     * @date 2024/10/9 13:34
     * @version 1.0
     */
    public void nettyFuture() throws Exception {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(3);
        EventLoop eventLoop = eventLoopGroup.next();
        io.netty.util.concurrent.Future<String> future = eventLoop.submit(() -> {
            System.out.println("异步任务开始执行"+"线程名"+Thread.currentThread().getName());
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "Success";
        });

        //同步
        /*String result = future.get();
        System.out.println("nettyFuture 同步阻塞 + " + result);*/
        //异步
        future.addListener(future1 -> System.out.println("nettyFuture 异步阻塞 + " + Thread.currentThread().getName() +"  " + future1.get()));
        eventLoopGroup.shutdownGracefully();
        //验证主线程不阻塞
        //jdkFuture();
    }

    private void nettyPromise() throws InterruptedException {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(2);
        EventLoop eventLoop = eventLoopGroup.next();
        final DefaultPromise<String> promise = new DefaultPromise<>(eventLoop);

        //给promise添加回调
        promise.addListener(new GenericFutureListener<io.netty.util.concurrent.Future<? super String>>() {
            @Override
            public void operationComplete(io.netty.util.concurrent.Future<? super String> future) throws Exception {
                if (future.isSuccess()){
                    System.out.println("nettyFuture任务执行成功");
                    System.out.println(future.get());
                } else {
                    throw new Exception("task submit fail");
                }
            }
        }).addListener(new GenericFutureListener<io.netty.util.concurrent.Future<? super String>>() {
            @Override
            public void operationComplete(io.netty.util.concurrent.Future<? super String> future) throws Exception {
                System.out.println("nettyFuture异步任务执行结束");
            }
        });

        eventLoopGroup.submit(()->{
            System.out.println("before promise set Success");
            try {
                Thread.sleep(1000);
                promise.setSuccess("Success");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        promise.sync();
        System.out.println("main 线程从 promise.sync() 阻塞中返回");

        eventLoopGroup.shutdownGracefully();
    }


}
